package cn.mjnxin.iot.base.process;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import cn.mjnxin.iot.base.concurrent.ThreadFactoryImpl;
import cn.mjnxin.iot.base.log.LogWorker;

/**
 * 同类数据处理总线
 * @author CLG 2023年11月30日
 * @version V1.0.0
 */
public abstract class DataProcessWorker<T> implements IWorker{
	private volatile boolean         active       = true;
	private final int                poolSize;
	private final ExecutorService    exService;
	/**
     * 内部消息总线
     */
    private final BlockingQueue<T>   dataQueue = new LinkedBlockingQueue<T>();
    
	public DataProcessWorker(int poolSize, String name) throws Exception {
		super();
		this.poolSize = poolSize;
		this.exService = Executors.newFixedThreadPool(poolSize, new ThreadFactoryImpl(name + "_Thread", false));
	}
	
	@Override
	public final void start() throws Exception {
		new Thread(this).start();
	}

    public final boolean add(T data) {
    	try {
    		dataQueue.put(data);
    		return true;
		} catch (InterruptedException e) {
			LogWorker.error("dataQueue.put(data) interrupted!", e);
		}
    	
    	return false;
    }

	// 清空缓存
	public void clear() {
		dataQueue.clear();
	}
	
	public final void run() {
		for (int i=0 ; i < poolSize ; i++ ){
			exService.execute(new Runnable() {
				@Override
				public void run() {
					while(active){
						T data = null;
						try {
							data = dataQueue.take();
	                    } catch (Exception e) {
	            			LogWorker.error("dataQueue.take() interrupted!", e);
	            			break;
	                    }
						if(data == null){
							continue;
						}

						try {
							proccess(data);
						} catch (Exception e) {
							LogWorker.error(e.getMessage(), e);
						}
						if (dataQueue.isEmpty()) {
							if (!active) {
								exService.shutdownNow();
							}
						}
                		else {
                			if (!active) {
                			// TODO:剩余消息没有消费完待处理
                			}
                		}
					}
				}
			});
		}
	}

	protected abstract void proccess(T data) throws Exception;

	public final void stopWorker() {
		active = false;
		dataQueue.clear();
	}
	
	public final boolean isShutdown() {
		if (dataQueue.isEmpty()) {
//			if (!Thread.currentThread().isInterrupted()) {
		    Thread.currentThread().interrupt(); // 中断take
			}
//		}
		return exService.isShutdown() && exService.isTerminated();
	}
}
